package belf.migrate.evaluate.helper;

import belf.migrate.api.job.Job;
import belf.migrate.api.model.CharacteristicInfo;
import belf.migrate.api.model.QualifiersModel;
import belf.migrate.api.table.schema.AbstractJdbcCatalog;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.taskconf.Task;
import belf.migrate.api.taskconf.TaskConf;
import belf.migrate.api.type.LevelType;
import belf.migrate.api.type.MatchingType;
import belf.migrate.core.jdbc.DBFunctionMapping;
import belf.migrate.core.jdbc.FunctionModel;
import belf.migrate.evaluate.model.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;

import java.util.*;

@Slf4j
public abstract class EvaluateBaseTool {

    protected JobContext jobContext;
    protected Job job;
    protected Task task;
    protected TaskConf taskConf;
    protected AbstractJdbcCatalog catalog;

    protected EvaluateBaseTool(JobContext jobContext, Job job) {
        this.job = job;
        this.jobContext = jobContext;
        this.task = job.getTask();
        this.taskConf = job.getTask().getTaskConf();
        this.catalog = jobContext.getSourceCatalog();
    }

    public QualifiersModel getSourceQualifiersModel() {

        return jobContext.getSourceQualifiersModel();
    }

    public QualifiersModel getSinkQualifiersModel() {
        return jobContext.getSinkQualifiersModel();
    }

    public abstract Object run();

    /**
     * 评估ddl语句中的函数差异
     * @param ddl
     * @param catalogs
     * @param taskType
     * @param matchingType
     * @param baseModel
     * @return
     */
    protected MatchingType functionDiff(String ddl, String[] catalogs, String taskType, MatchingType matchingType, EvaluateBaseModel baseModel, Map<String, Map<String, Integer>> functionRiskMap, String sinkCatalog) {
        //获取函数映射关系
        List<String> functionIncompatible = catalog.findFunctionIncompatible(
                ddl,
                catalogs[0],
                sinkCatalog);
        Map<String, FunctionModel> mapping = DBFunctionMapping.getInstance().getMapping(taskType);
        List<EvaluateFunctionModel> evaluateFunctionModels = new ArrayList<>();

        Map<String, String> map = new HashMap<>();
        //获取不兼容函数
        for (String funcName : functionIncompatible) {
            if (map.containsKey(funcName)) {
                continue;
            }
            map.put(funcName, funcName);
            EvaluateFunctionModel evaluateFunctionModel = new EvaluateFunctionModel();
            evaluateFunctionModel.setName(funcName);
            String level = mapping.get(funcName).getLevel();
            if (LevelType.WARNING.getName().equals(level) && matchingType.getId() < 1) {
                matchingType = MatchingType.CONVERT_MATCH;
            } else if (LevelType.ERROR.getName().equals(level) && matchingType.getId() < 2) {
                addMap(functionRiskMap, sinkCatalog, funcName.toUpperCase(Locale.ROOT));
                matchingType = MatchingType.NO_MATCH;
            }
            evaluateFunctionModel.setLevel(level);
            evaluateFunctionModel.setDesc(mapping.get(funcName).getDescription());
            evaluateFunctionModels.add(evaluateFunctionModel);
        }

        baseModel.setFuncDiff(evaluateFunctionModels);
        return matchingType;
    }

    /**
     * 评估ddl语句中的特性
     * @param baseModel
     * @param matchingType
     * @return
     */
    protected MatchingType characteristic(EvaluateBaseModel baseModel, MatchingType matchingType, List<CharacteristicInfo> characteristicInfo) {
        if (CollectionUtils.isNotEmpty(characteristicInfo)) {
            List<EvaluateCharacteristicModel> characteristicModelList = new ArrayList<>();
            for (CharacteristicInfo info : characteristicInfo) {
                EvaluateCharacteristicModel characteristicModel = new EvaluateCharacteristicModel();
                characteristicModel.setName(info.getName());
                characteristicModel.setLevel(LevelType.ERROR.getName());
                characteristicModel.setDesc(info.getDesc());
                characteristicModelList.add(characteristicModel);
            }
            baseModel.setCharacteristic(characteristicModelList);
            matchingType = MatchingType.NO_MATCH;
        } else if (matchingType == MatchingType.FULL_MATCH) {
            matchingType = MatchingType.CONVERT_MATCH;
        }
        return matchingType;
    }

    protected void addMap(Map<String, Map<String, Integer>> map, String catalog, String key) {
        Map<String, Integer> tempMap;
        if (!map.containsKey(catalog)) {
            tempMap = new HashMap<>();
            tempMap.put(key, 1);
            map.put(catalog, tempMap);
        } else {
            tempMap = map.get(catalog);
            if (tempMap.containsKey(key)) {
                tempMap.put(key, tempMap.get(key) + 1);
            } else {
                tempMap.put(key, 1);
            }
        }
        map.put(catalog, tempMap);
    }

    public List<EvaluateBaseModel> getBaseModelByMap(Map<String, List<EvaluateBaseModel>> baseModelMap, String key) {
        if (baseModelMap.containsKey(key)) {
            return baseModelMap.get(key);
        }
        return new ArrayList<>();
    }

    public EvaluateSinkResult getSinkResultList(Map<String, EvaluateSinkResult> sinkResultMap, String key) {
        if (sinkResultMap.containsKey(key)) {
            return sinkResultMap.get(key);
        } else {
            return new EvaluateSinkResult();
        }
    }
}
